package bjut.core.utils;

import bjut.core.enums.RpcConfigEnum;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.springframework.util.Assert;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @Author: zhuzhiming
 * @Date: 2022/4/7 17:05
 */
@Slf4j
public class CuratorUtil {

    private static final int SLEEP_TIME = 1000;
    private static final int MAX_RETRIES = 3;

    public static final String ZRPC_ROOT_PATH = "/zRpc/";
    private static final String DEFAULT_ZOOKEEPER_PATH = "127.0.0.1:2181";
    private static final Set<String> CREATED_PATH_SET = ConcurrentHashMap.newKeySet();
    private static final ConcurrentHashMap<String, List<String>> SERVICES_ADDRESS_MAP = new ConcurrentHashMap<>();
    private static CuratorFramework zkClient;


    public static CuratorFramework getZkClient() {

        if (zkClient != null && zkClient.getState() == CuratorFrameworkState.STARTED) {
            return zkClient;
        }

        String fileName = RpcConfigEnum.ZRPC_CONFIG_PATH.getPropertyValue();
        Properties properties = PropertiesFileUtil.readPropertiesFile(fileName);
        Assert.notNull(properties, "{} not exist" + RpcConfigEnum.ZRPC_CONFIG_PATH.getPropertyValue());
        String zkServerAddress = RpcConfigEnum.ZRPC_ZOOKEEPER_ADDRESS.getPropertyValue();
        String zkServer = properties.getProperty(zkServerAddress) == null ? DEFAULT_ZOOKEEPER_PATH : properties.getProperty(zkServerAddress);

        //重试策略
        ExponentialBackoffRetry backoffRetry = new ExponentialBackoffRetry(SLEEP_TIME, MAX_RETRIES);

        zkClient = CuratorFrameworkFactory.builder()
                .connectString(zkServer)
                .retryPolicy(backoffRetry)
                .build();

        zkClient.start();

        try {
            if (!zkClient.blockUntilConnected(30, TimeUnit.SECONDS)) {
                throw new RuntimeException("can not connect zk server");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return zkClient;
    }

    public static void createPersistentNode(CuratorFramework zkClient, String path) {

        try {
            if (CREATED_PATH_SET.contains(path) || zkClient.checkExists().forPath(path) != null) {
                log.info("the path [{}] already exist", path);
            } else {
                zkClient.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).forPath(path);
                log.info("the path [{}] create success", path);
            }
            CREATED_PATH_SET.add(path);
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage());
        }
    }


    public static List<String> getChildrenNodes(CuratorFramework zkClient, String rpcServiceName) {

        if (SERVICES_ADDRESS_MAP.containsKey(rpcServiceName)) {
            return SERVICES_ADDRESS_MAP.get(rpcServiceName);
        }

        String servicePath = ZRPC_ROOT_PATH + rpcServiceName;
        List<String> addressList;
        try {
            addressList = zkClient.getChildren().forPath(servicePath);
            SERVICES_ADDRESS_MAP.put(rpcServiceName, addressList);
            registerWatcher(zkClient, rpcServiceName);
        } catch (Exception e) {
            throw new RuntimeException("not find anyone service address");
        }
        return addressList;
    }

    @SneakyThrows
    private static void registerWatcher(CuratorFramework zkClient, String rpcServiceName) {

        String servicePath = ZRPC_ROOT_PATH + rpcServiceName;
        PathChildrenCache childrenCache = new PathChildrenCache(zkClient, servicePath, true);
        PathChildrenCacheListener listener = ((curatorFramework, pathChildrenCacheEvent) -> {
            List<String> serviceAddresses = curatorFramework.getChildren().forPath(servicePath);
            SERVICES_ADDRESS_MAP.put(rpcServiceName, serviceAddresses);
        });
        childrenCache.getListenable().addListener(listener);
        childrenCache.start();
    }


    public static void clearServices(CuratorFramework zkClient, InetSocketAddress inetSocketAddress) {

        CREATED_PATH_SET.stream().parallel().forEach(service -> {
            if (service.endsWith(inetSocketAddress.toString())) {
                try {
                    zkClient.delete().forPath(service);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

}
